package flink.examples.sql._03.source_sink.table.redis.mapper;

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommand;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisCommandDescription;
import org.apache.flink.streaming.connectors.redis.common.mapper.RedisMapper;
import org.apache.flink.table.data.RowData;

import java.nio.charset.StandardCharsets;


// SetRedisMapper需要进行修改，以接收并使用SerializationSchema
public class SetRedisMapper implements RedisMapper<RowData> {
    private final SerializationSchema<RowData> serializationSchema;

    public SetRedisMapper(SerializationSchema<RowData> serializationSchema) {
        this.serializationSchema = serializationSchema;
    }

    @Override
    public RedisCommandDescription getCommandDescription() {
        return new RedisCommandDescription(RedisCommand.SET, null);
    }

    @Override
    public String getKeyFromData(RowData data) {
        // Extract key from RowData
        // Assume the key is at position 0
        return data.getString(0).toString();
    }

    @Override
    public String getValueFromData(RowData data) {
        // Serialize value using serialization schema
        return new String(serializationSchema.serialize(data), StandardCharsets.UTF_8);
    }
}